// This file is part of OpenTSDB.
// Copyright (C) 2017 TeleDB Development Group.
//
// This program is free software: you can redistribute it and/or modify it
// under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 2.1 of the License, or (at your
// option) any later version.  This program is distributed in the hope that it
// will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty
// of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU Lesser
// General Public License for more details.  You should have received a copy
// of the GNU Lesser General Public License along with this program.  If not,
// see <http://www.gnu.org/licenses/>.
package net.opentsdb.tools.http;

import org.apache.http.HttpRequest;
import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpEntityEnclosingRequestBase;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.client.methods.HttpRequestBase;
import org.apache.http.concurrent.FutureCallback;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.Charset;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicInteger;

public class HttpSender {

    private static final Logger LOG = LoggerFactory.getLogger(HttpSender.class);
    private final CloseableHttpAsyncClient closeableHttpAsyncClient;

    public static final Charset DEFAULT_CHARSET = Charset.forName("UTF-8");
    private boolean httpCompress;
    private final AtomicInteger unCompletedTaskNum;

    public HttpSender(TSDBConfig config, CloseableHttpAsyncClient asyncClient) {
        this.closeableHttpAsyncClient = asyncClient;
        this.unCompletedTaskNum = new AtomicInteger(0);
    }

    public HttpResponse sendGetRequest(String url) {
        HttpGet httpGet = new HttpGet(url);
        httpGet.setHeader("accept", "application/json");
        return this.executeGetRequest(httpGet);
    }

    public HttpResponse postJson(String url, String json) {
        HttpPost httpPost = new HttpPost(url);
        return this.executeJson(httpPost, json);
    }

    private HttpResponse executeJson(HttpEntityEnclosingRequestBase request, String json) {
        if (json != null && json.length() > 0) {
            request.addHeader("Content-Type", "application/json");
            request.setEntity(this.generateStringEntity(json));
        }
        HttpResponse httpResponse = null;
        // LOG.info("request>>>" + request.toString());
        try {
            httpResponse = this.closeableHttpAsyncClient.execute(request, new FutureCallback<HttpResponse>() {
                @Override
                public void completed(HttpResponse httpResponse) {
                    // ignore
                }

                @Override
                public void failed(Exception e) {
                    LOG.error("send http request error", e.getMessage());
                }

                @Override
                public void cancelled() {

                }
            }).get();
        } catch (InterruptedException e) {
            LOG.error("", e.getMessage());
        } catch (ExecutionException e) {
            LOG.error("", e.getMessage());
        }
        return httpResponse;
    }

    private HttpResponse executeGetRequest(HttpRequestBase request) {
        request.addHeader("Content-Type", "application/json");
        HttpResponse httpResponse = null;
        LOG.info("request>>>" + request.toString());
        try {
            httpResponse = this.closeableHttpAsyncClient.execute(request, new FutureCallback<HttpResponse>() {
                @Override
                public void completed(HttpResponse httpResponse) {
                }

                @Override
                public void failed(Exception e) {
                    LOG.error("send http request error", e.getMessage());
                }

                @Override
                public void cancelled() {

                }
            }).get();
        } catch (InterruptedException e) {
            LOG.error("", e.getMessage());
        } catch (ExecutionException e) {
            LOG.error("", e.getMessage());
        }
        return httpResponse;
    }

    private StringEntity generateStringEntity(String json) {
        StringEntity stringEntity = new StringEntity(json, Charset.forName("UTF-8"));
        return stringEntity;
    }

    public void start() {
        this.closeableHttpAsyncClient.start();
    }

    public void close() {
        try {
            closeableHttpAsyncClient.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
